package org.example.mq.mqserver.datacenter;

/*
 * 消息管理类
 */

import lombok.extern.slf4j.Slf4j;
import org.example.mq.common.exception.MqException;
import org.example.mq.common.exception.QueueNotPersistentException;
import org.example.mq.common.tools.BinaryTool;
import org.example.mq.mqserver.core.Message;
import org.example.mq.mqserver.core.MessageQueue;
import org.springframework.stereotype.Component;

import java.io.*;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Arrays;
import java.util.LinkedList;

@Slf4j
@Component
public class MessageFileManager {
    //初始化后续在这添加
    public static void init() {

    }

    public static class Stat{
        //消息总数
        public int totalCount;

        //消息有效数量
        public int validCount;
    }


    // 获取队列文件目录
    private String getQueueDir(String queueName){
        return "./data/" + queueName;
    }

    //获取消息文件路径
    private String getQueueDataPath(String queueName){
        return getQueueDir(queueName) + "/message_data.bin";
    }

    //获取消息统计文件路径
    private String getQueueStatPath(String queueName){
        return getQueueDir(queueName) + "/message_stat.txt";
    }

    //获取复制消息文件目录
    private String getQueueNewDataPath(String queueName){
        return getQueueDir(queueName) + "/new_message_data.bin";
    }

    //读 stat 数据
    private Stat readStat(String queueName){
        Stat stat = new Stat();
        Path path = Paths.get(getQueueStatPath(queueName));

        try(BufferedReader bufferedReader = Files.newBufferedReader(path)) {
            stat.totalCount = Integer.parseInt(bufferedReader.readLine());
            stat.validCount = Integer.parseInt(bufferedReader.readLine());
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }

        return null;
    }

    //写入 stat 数据
    private void writeStat(String queueName,Stat stat){
        Path path = Paths.get(getQueueStatPath(queueName));

        try(BufferedWriter bufferedWriter = Files.newBufferedWriter(path)) {
            bufferedWriter.write(String.valueOf(stat.totalCount));
            bufferedWriter.newLine();
            bufferedWriter.write(String.valueOf(stat.validCount));
            bufferedWriter.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //创建对应目录
    public void createQueueFiles(String queueName) throws IOException {
        File baseDir = new File(getQueueDir(queueName));

        if(!baseDir.exists()){
            boolean res = baseDir.mkdirs();
            if (!res) {
                throw new IOException("目录创建失败！baseDir="+baseDir.getAbsolutePath());
            }
        }

        //创建对应文件
        File data = new File(baseDir+"/message_data.bin");
        File stat = new File(baseDir+"/message_stat.txt");

        if(!data.exists()){
            boolean created = data.createNewFile(); // 创建文件
            if (!created) {
                throw new IOException("创建文件失败！queueDataFile="+data.getAbsolutePath());
            }
        }

        if(!stat.exists()){
            boolean created = stat.createNewFile(); // 创建文件
            if (!created) {
                throw new IOException("创建文件失败！queueStatFile="+stat.getAbsolutePath());
            }
        }

        //对stat文件初始化
        Stat stat1 = new Stat();
        stat1.validCount=0;
        stat1.totalCount=0;
        writeStat(queueName,stat1);
    }

    //删除目录
    public void delQueueFiles(String queueName) throws IOException {
        File dataFile = new File(getQueueDataPath(queueName));
        File statFile = new File(getQueueStatPath(queueName));
        File baseDir = new File(getQueueDir(queueName));

        //先删除文件在删除目录
        boolean ok1 = dataFile.delete();
        boolean ok2 = statFile.delete();
        boolean ok3 = baseDir.delete();

        if(!ok1 || !ok2 || !ok3){
            throw new IOException("删除目录失败！queueDir:"+baseDir.getAbsolutePath());
        }
    }

    //检查目录是否存在
    public boolean checkQueueDirExist(String queueName){
        File queuePath = new File(getQueueDir(queueName));
        if(!queuePath.exists())return false;

        Path statFile = Paths.get(getQueueStatPath(queueName));
        Path dataFile = Paths.get(getQueueDataPath(queueName));

        if(!Files.exists(dataFile))return false;

        if(!Files.exists(statFile))return false;

        return true;
    }

    //消息持久化保存
    public void sendMessage(MessageQueue queue, Message message){
        //判断队列是否持久化,如果队列没有开启持久化就抛出
        if(!checkQueueDirExist(queue.getName())){
            throw new QueueNotPersistentException();
        }

        //将消息序列化
        byte[] messageBytes = BinaryTool.toBytes(message);
//        System.out.println("[messageBytes]:"+ Arrays.toString(messageBytes));

        //计算消息在文件中的起始位置
        //offsetBeg = 文件大小+4
        //offsetEnd = 文件大小+4+消息大小
        File messageDataFile = new File(getQueueDataPath(queue.getName()));
        message.setOffsetBeg(messageDataFile.length()+4);
        message.setOffsetEnd(messageDataFile.length()+4+messageBytes.length);

        //将消息写入文件中
        synchronized (queue) {
            try(OutputStream outputStream = new FileOutputStream(messageDataFile,true)) {
                BufferedOutputStream bufferedOutputStream = new BufferedOutputStream(outputStream);
                DataOutputStream dataOutputStream = new DataOutputStream(bufferedOutputStream);

                //先写入消息长度
                dataOutputStream.writeInt(messageBytes.length);
                //将序列化后的消息存入
                dataOutputStream.write(messageBytes);
                dataOutputStream.flush();
                //存入后更新stat文件内容
                Stat stat = readStat(queue.getName());
                stat.totalCount +=1;
                stat.validCount +=1;
                writeStat(queue.getName(),stat);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    //删除消息
    public void delMessage(MessageQueue queue,Message message) {
        //获取到消息,这里获取消息采用随机访问

        synchronized (queue) {
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "wr")) {
                byte[] getMessage = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];

                //设置读取光标起始位置
                randomAccessFile.seek(message.getOffsetBeg());

                randomAccessFile.read(getMessage);

                //反序列化
                Message objMessage = (Message) BinaryTool.toObject(getMessage);
                objMessage.setIsValid((byte) 0x0);

                byte[] saveMessage = BinaryTool.toBytes(objMessage);

                randomAccessFile.seek(message.getOffsetBeg());

                randomAccessFile.write(saveMessage);

                //更新stat
                Stat stat = readStat(queue.getName());
                stat.validCount -= 1;
                writeStat(queue.getName(), stat);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    //读取消息
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException {
        LinkedList<Message> messages = new LinkedList<>();
        File messageDataFile = new File(getQueueDataPath(queueName));

        try(InputStream inputStream = new FileInputStream(messageDataFile)){
            BufferedInputStream bufferedInputStream = new BufferedInputStream(inputStream);
            DataInputStream dataInputStream = new DataInputStream(bufferedInputStream);

            int currentCount = 0;

            while(true){
                //读取消息长度
                int messageSize = dataInputStream.readInt();

                byte[] messageBinary = new byte[messageSize];

                try {
                    dataInputStream.readFully(messageBinary); // 不用你自己判断长度了
                } catch (EOFException e) {
                    throw new MqException("[MessageFileManager],queueName:"+queueName+"消息数据不完整：期望读取 " + messageSize + " 字节，但文件已结束", e);
                }

                //反序列化
                Message message = (Message) BinaryTool.toObject(messageBinary);

                //检查消息是否有效
                if (!message.msgIsAlive()) {
                    currentCount += (4+messageSize);
                    continue;
                }

                message.setOffsetBeg(currentCount + 4);
                message.setOffsetEnd(currentCount + 4 + messageSize);
                currentCount+= (4+messageSize);
                messages.add(message);
            }
        }catch (EOFException eofException){
            //读取到文件末尾
            System.out.println("[MessageFileManage] 恢复 queueName: "+ queueName +",完成!");
        }

        return  messages;
    }

    public boolean checkGc(String queueName){
        //判断是否需要gc
        Stat stat = readStat(queueName);
        if(stat.totalCount<2000 || (double) stat.validCount /stat.totalCount > 0.5){
            return false;
        }
        return true;
    }


    //消息文件垃圾回收
    public void messageGc(MessageQueue queue) throws IOException {
        synchronized (queue){
            long gcBeg = System.currentTimeMillis();

            File newDataFile = new File(getQueueNewDataPath(queue.getName()));

            if(!newDataFile.exists()){
                boolean created = newDataFile.createNewFile(); // 创建文件
                if (!created) {
                    throw new IOException("创建文件失败！newDataFile="+newDataFile.getAbsolutePath());
                }
            }

            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());

            //写入文件
            OutputStream outputStream = new FileOutputStream(newDataFile,true);
            BufferedOutputStream bufferedOutputStream = new BufferedOutputStream(outputStream);
            DataOutputStream dataOutputStream = new DataOutputStream(bufferedOutputStream);

            messages.forEach(message -> {
                byte[] messageBinary = BinaryTool.toBytes(message);
                try {
                    dataOutputStream.writeInt(messageBinary.length);
                    dataOutputStream.write(messageBinary);
                } catch (IOException e) {
                    throw new MqException("queue:"+queue.getName()+"[messageGc]发生错误");
                }
            });

            //删除旧文件
            File oldDataFile = new File(getQueueDataPath(queue.getName()));
            boolean ok = oldDataFile.delete();

            if(!ok){
                throw new MqException("queue:"+ queue.getName() +"[messageGc]删除旧文件失败！");
            }

            ok = newDataFile.renameTo(oldDataFile);
            if(!ok){
                throw new MqException("queue:"+ queue.getName() +"[messageGc],重命名文件失败！");
            }

            //更新统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount=messages.size();
            stat.validCount=messages.size();

            writeStat(queue.getName(),stat);
            long gcEnd = System.currentTimeMillis();

            System.out.println("[queue:"+queue.getName()+ "]gc耗时:"+ (gcEnd-gcBeg) + "ms");
        }
    }
}
